package com.dongzili.binlog.common.config.redis;

import cn.hutool.json.JSONUtil;
import com.dongzili.binlog.common.binlogsynchronization.BinLogConstants;
import com.dongzili.binlog.common.binlogsynchronization.BinLogItem;
import com.dongzili.binlog.common.config.es.EsUtils;
import com.dongzili.binlog.custom.entity.SBrowseLogModel;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.shyiko.mysql.binlog.event.EventType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.Map;

@Slf4j
@Component
public class RedisStreamConsumer implements StreamListener<String, ObjectRecord<String, BinLogItem>>{

    @Resource
    RedisTemplate<String, BinLogItem> redisTemplate;

    @Override
    public void onMessage(ObjectRecord<String, BinLogItem> message) {

        log.info("消息消费...");
        String stream = message.getStream();
        RecordId id = message.getId();
        BinLogItem value = message.getValue();
        log.info("[自动ack]  接收到一个消息 stream:[{}],id:[{}],value: {}", stream, id, value);
        Map<String, Serializable> data = value.getAfter();

        ObjectMapper mapper = new ObjectMapper();
        SBrowseLogModel model = null;
        try {
            model = mapper.readValue(JSONUtil.toJsonStr(data), SBrowseLogModel.class);
            log.info("model: {}",model);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        if(EventType.isWrite(value.getEventType())){
            //TODO 处理binlog
            log.info("新增数据: {}",JSONUtil.toJsonStr(data));
            EsUtils.getMapper(SBrowseLogModel.class).insert(model);
        }
        if(EventType.isDelete(value.getEventType())){
            //TODO 处理binlog
            log.info("删除数据: {}",JSONUtil.toJsonStr(data));
            EsUtils.getMapper(SBrowseLogModel.class).deleteById(model.getId());
        }
        if(EventType.isUpdate(value.getEventType())){
            //TODO 处理binlog
            log.info("修改数据: {}",JSONUtil.toJsonStr(data));
            EsUtils.getMapper(SBrowseLogModel.class).updateById(model);
        }


        assert stream != null;
        redisTemplate.opsForStream().acknowledge(stream, StreamMessageListenerConfig.streamKey, id.getValue());
        redisTemplate.opsForStream().delete(stream, id.getValue());

    }
}